package com.swak.reactivex.transport;

import static com.swak.Constants.Default_Timeout;
import static com.swak.Constants.Timeout;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.swak.exception.RemotingException;
import com.swak.exception.TimeoutException;
import com.swak.reactivex.threads.SwakThreadFactory;
import com.swak.timer.HashedWheelTimer;
import com.swak.timer.Timeout;
import com.swak.timer.Timer;
import com.swak.timer.TimerTask;

class TimeoutFuture extends CompletableFuture<Object> {

	private static final Logger logger = LoggerFactory.getLogger(TimeoutFuture.class);
	private static final Map<Long, Channel> CHANNELS = new ConcurrentHashMap<>();
	private static final Map<Long, TimeoutFuture> FUTURES = new ConcurrentHashMap<>();
	public static final Timer TIME_OUT_TIMER = new HashedWheelTimer(
			new SwakThreadFactory("Rpc.Future-timeout-", true, new AtomicInteger(0)), 30, TimeUnit.MILLISECONDS);

	// invoke id.
	private final Long id;
	private final Channel channel;
	private final Message request;
	private final int timeout;
	private final long start = System.currentTimeMillis();
	private volatile long sent;
	private Timeout timeoutCheckTask;

	private TimeoutFuture(Channel channel, Message request, int timeout) {
		this.channel = channel;
		this.request = request;
		this.id = request.getId();
		this.timeout = timeout > 0 ? timeout : channel.url().getIntParameter(Timeout, Default_Timeout);
		// put into waiting map.
		FUTURES.put(id, this);
		CHANNELS.put(id, channel);
	}

	/**
	 * check time out of the future
	 */
	private static void timeoutCheck(TimeoutFuture future) {
		TimeoutCheckTask task = new TimeoutCheckTask(future.getId());
		future.timeoutCheckTask = TIME_OUT_TIMER.newTimeout(task, future.getTimeout(), TimeUnit.MILLISECONDS);
	}

	/**
	 * init a DefaultFuture 1.init a DefaultFuture 2.timeout check
	 *
	 * @param channel channel
	 * @param request the request
	 * @param timeout timeout
	 * @return a new DefaultFuture
	 */
	public static TimeoutFuture newFuture(Channel channel, Message request, int timeout) {
		final TimeoutFuture future = new TimeoutFuture(channel, request, timeout);
		// timeout check
		timeoutCheck(future);
		return future;
	}

	public static TimeoutFuture getFuture(long id) {
		return FUTURES.get(id);
	}

	public static boolean hasFuture(Channel channel) {
		return CHANNELS.containsValue(channel);
	}

	public static void sent(Channel channel, Message request) {
		TimeoutFuture future = FUTURES.get(request.getId());
		if (future != null) {
			future.doSent();
		}
	}

	/**
	 * close a channel when a channel is inactive directly return the unfinished
	 * requests.
	 *
	 * @param channel channel to close
	 */
	public static void closeChannel(Channel channel) {
		for (Map.Entry<Long, Channel> entry : CHANNELS.entrySet()) {
			if (channel.equals(entry.getValue())) {
				TimeoutFuture future = getFuture(entry.getKey());
				if (future != null && !future.isDone()) {
					Message disconnectResponse = Message.forResponse(future.request);
					disconnectResponse.setStatus(Message.CHANNEL_INACTIVE);
					disconnectResponse.setData("Channel " + channel
							+ " is inactive. Directly return the unFinished request : " + future.getRequest());
					TimeoutFuture.received(channel, disconnectResponse);
				}
			}
		}
	}

	public static void received(Channel channel, Message response) {
		received(channel, response, false);
	}

	public static void received(Channel channel, Message response, boolean timeout) {
		try {
			TimeoutFuture future = FUTURES.remove(response.getId());
			if (future != null) {
				Timeout t = future.timeoutCheckTask;
				if (!timeout) {
					// decrease Time
					t.cancel();
				}
				future.doReceived(response);
			} else {
				logger.warn("The timeout response finally returned at "
						+ (new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(new Date())) + ", response status is "
						+ response.getStatus()
						+ (channel == null ? ""
								: ", channel: " + channel.localAddress() + " -> " + channel.remoteAddress())
						+ ", please check provider side for detailed result.");
			}
		} finally {
			CHANNELS.remove(response.getId());
		}
	}

	@Override
	public boolean cancel(boolean mayInterruptIfRunning) {
		Message errorResult = Message.forResponse(this.request);
		errorResult.setStatus(Message.CLIENT_ERROR);
		errorResult.setData("request future has been canceled.");
		this.doReceived(errorResult);
		FUTURES.remove(id);
		CHANNELS.remove(id);
		return true;
	}

	public void cancel() {
		this.cancel(true);
	}

	private void doReceived(Message res) {
		if (res == null) {
			throw new IllegalStateException("response cannot be null");
		}
		try {
			if (res.getStatus() == Message.OK) {
				this.complete(res);
			} else if (res.getStatus() == Message.CLIENT_TIMEOUT || res.getStatus() == Message.SERVER_TIMEOUT) {
				this.completeExceptionally(
						new TimeoutException(res.getStatus() == Message.SERVER_TIMEOUT, res.getError()));
			} else {
				this.completeExceptionally(new RemotingException(res.getError()));
			}
		} catch (Throwable e) {
			this.completeExceptionally(e);
		}
	}

	private long getId() {
		return id;
	}

	private Channel getChannel() {
		return channel;
	}

	private boolean isSent() {
		return sent > 0;
	}

	public Message getRequest() {
		return request;
	}

	private int getTimeout() {
		return timeout;
	}

	private void doSent() {
		sent = System.currentTimeMillis();
	}

	private String getTimeoutMessage(boolean scan) {
		long nowTimestamp = System.currentTimeMillis();
		return (sent > 0 ? "Waiting server-side response timeout" : "Sending request timeout in client-side")
				+ (scan ? " by scan timer" : "") + ". start time: "
				+ (new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(new Date(start))) + ", end time: "
				+ (new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(new Date(nowTimestamp))) + ","
				+ (sent > 0 ? " client elapsed: " + (sent - start) + " ms, server elapsed: " + (nowTimestamp - sent)
						: " elapsed: " + (nowTimestamp - start))
				+ " ms, timeout: " + timeout + " ms, request: "
				+ (logger.isDebugEnabled() ? request : getRequestWithoutData()) + ", channel: " + channel.localAddress()
				+ " -> " + channel.remoteAddress();
	}

	private Message getRequestWithoutData() {
		Message newRequest = request;
		newRequest.setData(null);
		return newRequest;
	}

	private static class TimeoutCheckTask implements TimerTask {

		private final Long requestID;

		TimeoutCheckTask(Long requestID) {
			this.requestID = requestID;
		}

		@Override
		public void run(Timeout timeout) {
			TimeoutFuture future = TimeoutFuture.getFuture(requestID);
			if (future == null || future.isDone()) {
				return;
			}
			notifyTimeout(future);
		}

		private void notifyTimeout(TimeoutFuture future) {
			// create exception response.
			Message timeoutResponse = Message.forResponse(future.request);
			// set timeout status.
			timeoutResponse.setStatus(future.isSent() ? Message.SERVER_TIMEOUT : Message.CLIENT_TIMEOUT);
			timeoutResponse.setData(future.getTimeoutMessage(true));
			// handle response.
			TimeoutFuture.received(future.getChannel(), timeoutResponse, true);
		}
	}
}